/*
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

/*
 *
 *
 *
 *
 *
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;

/**
 * A {@link CompletionService} that uses a supplied {@link Executor}
 * to execute tasks.  This class arranges that submitted tasks are,
 * upon completion, placed on a queue accessible using {@code take}.
 * The class is lightweight enough to be suitable for transient use
 * when processing groups of tasks.
 *
 * <p>
 *
 * <b>Usage Examples.</b>
 * <p>
 * Suppose you have a set of solvers for a certain problem, each
 * returning a value of some type {@code Result}, and would like to
 * run them concurrently, processing the results of each of them that
 * return a non-null value, in some method {@code use(Result r)}. You
 * could write this as:
 *
 * <pre> {@code
 * void solve(Executor e,
 *            Collection<Callable<Result>> solvers)
 *     throws InterruptedException, ExecutionException {
 *     CompletionService<Result> ecs
 *         = new ExecutorCompletionService<Result>(e);
 *     for (Callable<Result> s : solvers)
 *         ecs.submit(s);
 *     int n = solvers.size();
 *     for (int i = 0; i < n; ++i) {
 *         Result r = ecs.take().get();
 *         if (r != null)
 *             use(r);
 *     }
 * }}</pre>
 * <p>
 * Suppose instead that you would like to use the first non-null result
 * of the set of tasks, ignoring any that encounter exceptions,
 * and cancelling all other tasks when the first one is ready:
 *
 * <pre> {@code
 * void solve(Executor e,
 *            Collection<Callable<Result>> solvers)
 *     throws InterruptedException {
 *     CompletionService<Result> ecs
 *         = new ExecutorCompletionService<Result>(e);
 *     int n = solvers.size();
 *     List<Future<Result>> futures
 *         = new ArrayList<Future<Result>>(n);
 *     Result result = null;
 *     try {
 *         for (Callable<Result> s : solvers)
 *             futures.add(ecs.submit(s));
 *         for (int i = 0; i < n; ++i) {
 *             try {
 *                 Result r = ecs.take().get();
 *                 if (r != null) {
 *                     result = r;
 *                     break;
 *                 }
 *             } catch (ExecutionException ignore) {}
 *         }
 *     }
 *     finally {
 *         for (Future<Result> f : futures)
 *             f.cancel(true);
 *     }
 *
 *     if (result != null)
 *         use(result);
 * }}</pre>
 *
 * <pre>
 * 1- 先简单说明一下：对于 {@link java.util.concurrent.Future#get()} 方法，我们知道当调用 get() 方法获取子线程的执行结果时，程序会【阻塞】在那里，直至获取返回结果。
 * 2- 从而引出的问题：
 *      1). 现有代码如下。假设 任务 A 由于参数原因，执行时间相对 任务 B、C、D 都要长很多，但是按照程序的执行顺序，程序在 get() 任务 A 的执行结果时会阻塞在那里，
 *          那么就会导致 B、C、D 【各自】的【后续】任务没办法执行，当然，在 get方法获取任务 A 结果的时候 B、C、D 三个任务可能已经执行完成了，就等着来获取结果了。
 *
 *      ```插曲
 *          【那么就会导致 B、C、D 【各自】的【后续】任务没办法执行】这句话是什么意思呢？
 *          这么理解，A、B、C、D 四个任务在执行完成之后，需要执行不同的逻辑代码。
 *
 *          TODO 那么，在这里我本人有一个问题，既然 四个任务在执行完成之后，还是需要执行【不同】的逻辑代码，那为何不把他们各自的后续任务就放在原本的任务基础之上去做呢。搞不懂。
 *      ```
 *
 *      又因为每个任务的执行时间是不固定的，所以无论怎样调整将任务放到 futureList 的顺序，都不合适，这就是 Future#get() 方法带来的致命问题。
 *
 *      2). 当然，① B、C、D 如果没有什么【各自后续】任务的话，就无所谓了；② 如果程序确实是要等到 A、B、C、D 四个任务都执行完成后才能继续向下执行公共逻辑的话，那也就无所谓了。
 *
 *      @<code>
 *          ExecutorService executorService = Executors.newFixedThreadPool(4);
 *
 *          List<Future> futureList = new ArrayList<Future<Integer>>();
 *          futureList.add(executorService.submit(A));
 *          futureList.add(executorService.submit(B));
 *          futureList.add(executorService.submit(C));
 *          futureList.add(executorService.submit(D));
 *
 *          // 遍历 futureList，通过 get() 方法获取每个 future 的执行结果。
 *          for (Future future : futureList) {
 *              Integer result = future.get();
 *              log.info("result: {}", result);
 *          }
 *      </code>
 *
 * 3- 由于上面所存在的问题，SO，{@link ExecutorCompletionService} 这个家伙也就应运而生了。
 *    1). 目的：ExecutorCompletionService 可以获取到最先执行完成的任务的结果。
 *        如果你使用过消息队列，那么 ExecutorCompletionService 就很容易理解了。
 *    2). 实现大体逻辑：哪个任务执行完成，就直接把执行结果放到【队列】中，这样消费者拿到的结果自然就是最早完成的那个了。
 *        有【任务】，有相应的【结果队列】。
 *        ① 既然是异步任务，那就可能要用到 Runnable 或者 Callable；
 *        ② 既然能获取到结果，那就会用到 Future 了。
 * 4- 要想彻底理解 ExecutorCompletionService，我们只需要知道一个问题的答案就可以了：
 *    它是如何将异步任务结果放到 {@link ExecutorCompletionService#completionQueue} 这个阻塞队列中的？
 *    见 {@link QueueingFuture#done()}
 * </pre>
 */
public class ExecutorCompletionService<V> implements CompletionService<V> {
    /**
     * 从这里可以看得出来，异步任务的执行还是得交给 {@link Executor}（线程池）来完成。
     */
    private final Executor executor;
    private final AbstractExecutorService aes;

    /**
     * 这个【阻塞队列】是用来存放【异步任务的执行结果 future】
     */
    private final BlockingQueue<Future<V>> completionQueue;

    /**
     * FutureTask extension to enqueue upon completion
     */
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }

        /**
         * <pre>
         * 1). 这个方法在 {@link FutureTask#finishCompletion()} 中会被调用。
         * 2). 这个 {@link FutureTask#finishCompletion()} 方法的 调用方法有两个
         *      {@link FutureTask#set(Object)} 和 {@link FutureTask#setException(Throwable)}。
         *      而这两个方法在 {@link FutureTask#run()} 中又会被调用，用来存放异步任务的执行结果（正常结果 或 Exception）。
         *
         * 最重要的还是在 {@link FutureTask#finishCompletion()} 这个方法中，这个方法会来调用当前的这个 done() 方法，
         * 这里其实就是一个【回调】。
         * </pre>
         */
        protected void done() {
            /* 将 task 放到【阻塞队列】中。
             * 当方法执行到这里的时候，就已经表明，此时的 task 已经是前序步骤 set 过的结果的 task 了（即，已经执行过 FutureTask#run() 方法了）。
             * 最终，completionQueue队列中的任务都是按照执行完成的先后顺序存储的。
             */
            completionQueue.add(task);
        }

        private final Future<V> task;
    }

    private RunnableFuture<V> newTaskFor(Callable<V> task) {
        if (aes == null)
            return new FutureTask<V>(task);
        else
            return aes.newTaskFor(task);
    }

    private RunnableFuture<V> newTaskFor(Runnable task, V result) {
        if (aes == null)
            return new FutureTask<V>(task, result);
        else
            return aes.newTaskFor(task, result);
    }

    /**
     * Creates an ExecutorCompletionService using the supplied
     * executor for base task execution and a
     * {@link LinkedBlockingQueue} as a completion queue.
     *
     * @param executor the executor to use
     * @throws NullPointerException if executor is {@code null}
     */
    public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
                (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }

    /**
     * Creates an ExecutorCompletionService using the supplied
     * executor for base task execution and the supplied queue as its
     * completion queue.
     *
     * @param executor        the executor to use
     * @param completionQueue the queue to use as the completion queue
     *                        normally one dedicated for use by this service. This
     *                        queue is treated as unbounded -- failed attempted
     *                        {@code Queue.add} operations for completed tasks cause
     *                        them not to be retrievable.
     * @throws NullPointerException if executor or completionQueue are {@code null}
     */
    public ExecutorCompletionService(Executor executor,
                                     BlockingQueue<Future<V>> completionQueue) {
        if (executor == null || completionQueue == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
                (AbstractExecutorService) executor : null;
        this.completionQueue = completionQueue;
    }

    public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture(f));
        return f;
    }

    public Future<V> submit(Runnable task, V result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task, result);
        executor.execute(new QueueingFuture(f));
        return f;
    }

    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }

    public Future<V> poll() {
        return completionQueue.poll();
    }

    public Future<V> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
        return completionQueue.poll(timeout, unit);
    }

}
